package com.issac.messagequeue.mqserver;

import com.issac.messagequeue.mqserver.common.Consumer;
import com.issac.messagequeue.mqserver.common.MqException;
import com.issac.messagequeue.mqserver.coreclass.*;
import com.issac.messagequeue.mqserver.datacenter.DiskDataCenter;
import com.issac.messagequeue.mqserver.datacenter.MemoryDataCenter;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 表示一个虚拟主机：管理自己的交换机、队列、绑定、消息数据
 * 提供api给上层调用
 */
@Data
@Slf4j
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();

    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    private final Object exchangeLocker = new Object();
    private final Object msgQueueLocker = new Object();

    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;

        diskDataCenter.init();

        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            log.error("[VirtualHost] MemoryDataCenter.recovery 恢复内存失败！");
        }
    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        // 交换机名 = 虚拟主机名 + 交换机名
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 通过内存查询交换机是否已存在
                Exchange searchResultFromMDC = memoryDataCenter.getExchange(exchangeName);
                if (searchResultFromMDC != null) {
                    log.info("[VirtualHost] 该交换机已存在！ exchangeName=" + exchangeName);
                    return true;
                }
                // 2. 创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 3. 交换机数据写入硬盘
                if (durable) {
                    diskDataCenter.addExchange(exchange);
                    log.info("[VirtualHost] 交换机写入硬盘成功！ exchangeName=" + exchangeName);
                }
                // 4. 交换机数据写入内存
                memoryDataCenter.addExchange(exchange);
                log.info("[VirtualHost] 交换机创建完成！ exchange：" + exchange);
            }
            return true;
        } catch (Exception e) {
            log.error("[VirtualHost] 交换机创建失败！ exchangeName" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean exchangeDelete(String exchangeName) throws MqException {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 找到交换机
                Exchange needToDelete = memoryDataCenter.getExchange(exchangeName);
                if (needToDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在！无法删除！");
                }
                // 2. 删除硬盘数据
                if (needToDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                log.info("[VirtualHost] 交换机已从硬盘中删除！ exchangeName=" + exchangeName);
                // 3. 删除内存数据
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("[VirtualHost] 交换机已从内存中删除！ exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            log.error("\"[VirtualHost] 交换机删除失败！ exchangeName" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean msgQueueDeclare(String msgQueueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        msgQueueName = virtualHostName + msgQueueName;
        try {
            synchronized (msgQueueLocker) {
                MSGQueue searchResultFromMDC = memoryDataCenter.getMsgQueue(msgQueueName);
                if (searchResultFromMDC != null) {
                    log.info("[VirtualHost] 该消息队列已存在！ msgQueueName=" + msgQueueName);
                    return true;
                }

                MSGQueue msgQueue = new MSGQueue();
                msgQueue.setName(msgQueueName);
                msgQueue.setDurable(durable);
                msgQueue.setExclusive(exclusive);
                msgQueue.setAutoDelete(autoDelete);
                msgQueue.setArguments(arguments);

                if (durable) {
                    diskDataCenter.addMSGQueue(msgQueue);
                    log.info("[VirtualHost] 消息队列写入硬盘成功！ msgQueueName=" + msgQueueName);
                }

                memoryDataCenter.addMSGQueue(msgQueue);
                log.info("[VirtualHost] 消息队列写入内存成功！ msgQueueName=" + msgQueueName);
            }
            return true;
        } catch (Exception e) {
            log.error("[VirtualHost] 消息队列创建失败！ msgQueueName=" + msgQueueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean msgQueueDelete(String msgQueueName) {
        msgQueueName = virtualHostName + msgQueueName;
        try {
            synchronized (msgQueueLocker) {
                // 1. 找到交换机
                MSGQueue needToDelete = memoryDataCenter.getMsgQueue(msgQueueName);
                if (needToDelete == null) {
                    throw new MqException("[VirtualHost] 消息队列不存在！无法删除！");
                }
                // 2. 删除硬盘数据
                if (needToDelete.isDurable()) {
                    diskDataCenter.deleteMSGQueue(msgQueueName);
                }
                log.info("[VirtualHost] 消息队列已从硬盘中删除！ msgQueueName=" + msgQueueName);
                // 3. 删除内存数据
                memoryDataCenter.deleteMsgQueue(msgQueueName);
                log.info("[VirtualHost] 消息队列已从内存中删除！ msgQueueName=" + msgQueueName);
            }
            return true;
        } catch (Exception e) {
            log.error("[VirtualHost] 消息队列删除失败！ msgQueueName" + msgQueueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean msgQueueBind(String msgQueueName, String exchangeName, String bindingKey) {
        msgQueueName = virtualHostName + msgQueueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (msgQueueLocker) {
                    Binding searchResultFromMDC = memoryDataCenter.getSpecifiedBinding(exchangeName, msgQueueName);
                    if (searchResultFromMDC != null) {
                        throw new MqException("[VirtualHost] 绑定已存在！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
                    }

                    // 验证bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey非法！ bindingKey=" + bindingKey);
                    }

                    Binding binding = new Binding();
                    binding.setQueueName(msgQueueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);

                    // 检查对应的交换机和队列是否存在
                    MSGQueue msgQueueFromMDC = memoryDataCenter.getMsgQueue(msgQueueName);
                    if (msgQueueFromMDC == null) {
                        throw new MqException("[VirtualHost] 消息队列不存在！ msgQueueName=" + msgQueueName);
                    }
                    Exchange exchangeFromMDC = memoryDataCenter.getExchange(exchangeName);
                    if (exchangeFromMDC == null) {
                        throw new MqException("[VirtualHost] 交换机不存在！ exchangeName=" + exchangeName);
                    }

                    // 写入：硬盘、内存
                    if (msgQueueFromMDC.isDurable() && exchangeFromMDC.isDurable()) {
                        diskDataCenter.addBinding(binding);
                    }
                    log.info("[VirtualHost] 绑定写入硬盘成功！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
                    memoryDataCenter.addBingding(binding);
                    log.info("[VirtualHost] 绑定写入内存成功！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
                }
            }
            log.info("[VirtualHost] 绑定创建成功！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
            return true;
        } catch (MqException e) {
            log.error("[VirtualHost] 绑定创建失败！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean msgQueueUnbind(String msgQueueName, String exchangeName) {
        msgQueueName = virtualHostName + msgQueueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (msgQueueLocker) {
                    // 验证binding是否存在
                    Binding bindingFromMDC = memoryDataCenter.getSpecifiedBinding(exchangeName, msgQueueName);
                    if (bindingFromMDC == null) {
                        throw new MqException("[VirtualHost] 绑定不存在！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
                    }

                    diskDataCenter.deleteBinding(bindingFromMDC);
                    log.info("[VirtualHost] 绑定从硬盘删除成功！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
                    memoryDataCenter.deleteBinding(bindingFromMDC);
                    log.info("[VirtualHost] 绑定从内存删除成功！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("[VirtualHost] 绑定从硬盘删除失败！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息到指定交换机/队列
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            exchangeName = virtualHostName + exchangeName;

            // 检查routingKey是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法！ routingKey=" + routingKey);
            }

            // 查看指定交换机是否存在
            Exchange exchangeFromMDC = memoryDataCenter.getExchange(exchangeName);
            if (exchangeFromMDC == null) {
                throw new MqException("[VirtualHost] 交换机不存在！ exchangeName=" + exchangeName);
            }

            // 根据交换机类型发送消息
            if (exchangeFromMDC.getType() == ExchangeType.DIRECT) {
                // 直接交换机
                // 以routingKey作为队列名字
                // 无视绑定关系
                String msgQueueName = virtualHostName + routingKey;

                MSGQueue msgQueueFromMDC = memoryDataCenter.getMsgQueue(msgQueueName);
                if (msgQueueFromMDC == null) {
                    throw new MqException("[VirtualHost] 指定消息队列不存在！ msgQueueName=" + msgQueueName);
                }

                Message message = Message.createMessageWithId(basicProperties, body, routingKey);
                sendMessage(msgQueueFromMDC, message);
                log.info("[VirtualHost] 消息发送到指定队列成功！ exchangeName=" + exchangeName + ", msgQueueName=" + msgQueueName);
            } else {
                // fanout 和 topic
                // 1)  获取绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 获取绑定对象
                    Binding binding = entry.getValue();

                    // 判断队列是否存在
                    MSGQueue msgQueueFromMDC = memoryDataCenter.getMsgQueue(binding.getQueueName());
                    if (msgQueueFromMDC == null) {
                        log.error("[VirtualHost] basicPublish 发送消息时，发现消息队列不存在！ msgQueueName=" + binding.getQueueName());
                        // 这里不抛出异常，为了不影响其他队列的消息传输
                        continue;
                    }

                    // 2） 构造消息对象
                    Message message = Message.createMessageWithId(basicProperties, body, routingKey);

                    // 3） 判定这个消息能否转发给该队列
                    if (!router.route(exchangeFromMDC.getType(), binding, message)) {
                        log.error("[VirtualHost] basicPublish 发送消息时，当前队列不符合要求，继续匹配......");
                        continue;
                    }

                    // 4)  转发给队列
                    sendMessage(msgQueueFromMDC, message);
                    log.info("[VirtualHost] basicPublish 消息发送到队列成功！ msgQueue=" + msgQueueFromMDC + ", messageId=" + message.getMessageId());
                }
            }
            return true;
        } catch (Exception e) {
            log.error("[VirtualHost] basicPublish 消息发送到队列失败！");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue msgQueue, Message message) throws IOException, MqException, InterruptedException {
        // 这里发送消息就是：把消息写到硬盘、内存
        int deliverMode = message.getDeliverMode();
        if (deliverMode == 2) {
            // 需要持久化
            diskDataCenter.sendMessage(message, msgQueue);
        }
        log.info("[VirtualHost] 消息写入硬盘成功！");
        // 消息写入内存
        memoryDataCenter.sendMessage(message, msgQueue);
        log.info("[VirtualHost] 消息写入内存成功！");

        // 通知消费者可以消费
        consumerManager.notifyConsume(msgQueue.getName());
    }

    /**
     * 订阅消息
     * 添加一个队列的订阅者，当队列收到收到消息后，把消息推送给对应订阅者
     *
     * @Param consumerTag: 消费者的身份标识
     * @Param autoAck: 消息被消费完成后，true-自动应答、false-手动应答
     * @Param consumer: 回调函数，类型为函数式接口。后续调用 basicConsume 可以直接使用lambda表达式
     */
    public boolean basicConsume(String consumerTag, String msgQueueName, boolean autoAck, Consumer consumer) {
        msgQueueName = virtualHostName + msgQueueName;
        try {
            consumerManager.addConsumer(consumerTag, msgQueueName, autoAck, consumer);
            log.info("[VirtualHost] basicConsume 消息订阅成功！ msgQueueName=" + msgQueueName);
            return true;
        } catch (Exception e) {
            log.error("VirtualHost] basicConsume 消息订阅失败！ msgQueueName=" + msgQueueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicAck(String msgQueueName, String messageId) {
        msgQueueName = virtualHostName + msgQueueName;
        try {
            // 1. 获取消息和队列
            Message message = memoryDataCenter.getMessageById(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在！ messageId=" + messageId);
            }
            MSGQueue msgQueue = memoryDataCenter.getMsgQueue(msgQueueName);
            if (msgQueue == null) {
                throw new MqException("[VirtualHost] 要确认的消息队列不存在！ msgQueueName=" + msgQueueName);
            }
            // 2. 删除硬盘数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(message, msgQueue);
            }
            // 3. 删除消息中心数据
            memoryDataCenter.deleteMessageById(messageId);
            // 4. 删除待确认集合数据
            memoryDataCenter.deleteMessageWaitAck(msgQueueName, messageId);
            log.info("[VirtualHost] basicAck 成功！ 消息被成功确认！ msgQueueName=" + msgQueueName + ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            log.error("[VirtualHost] basicAck 失败！ 消息确认失败！ msgQueueName=" + msgQueueName + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
